<!DOCTYPE html>
<html lang="zh-cn">
<head>
    
    <link type="text/css" rel="stylesheet" href="/bundles/blog-common.css?v=KOZafwuaDasEedEenI5aTy8aXH0epbm6VUJ0v3vsT_Q1"/>
<link id="MainCss" type="text/css" rel="stylesheet" href="/skins/ThinkInside/bundle-ThinkInside.css?v=RRjf6pEarGnbXZ86qxNycPfQivwSKWRa4heYLB15rVE1"/>
<link type="text/css" rel="stylesheet" href="/blog/customcss/428549.css?v=%2fam3bBTkW5NBWhBE%2fD0lcyJv5UM%3d"/>

</head>
<body>
<a name="top"></a>

<div id="page_begin_html"></div><script>load_page_begin_html();</script>

<div id="topics">
	<div class = "post">
		<h1 class = "postTitle">
			<a id="cb_post_title_url" class="postTitle2" href="https://www.cnblogs.com/frankdeng/p/9301688.html">Spark（十一）Spark分区</a>
		</h1>
		<div class="clear"></div>
		<div class="postBody">
			<div id="cnblogs_post_body" class="blogpost-body"><h2>一、分区的概念</h2>
<p>　　分区是RDD内部并行计算的一个计算单元，RDD的数据集在逻辑上被划分为多个分片，每一个分片称为分区，分区的格式决定了并行计算的粒度，而每个分区的数值计算都是在一个任务中进行的，因此任务的个数，也是由RDD(准确来说是作业最后一个RDD)的分区数决定。</p>
<h2>二、为什么要进行分区</h2>
<p>　　数据分区，在分布式集群里，网络通信的代价很大，减少网络传输可以极大提升性能。mapreduce框架的性能开支主要在io和网络传输，io因为要大量读写文件，它是不可避免的，但是网络传输是可以避免的，把大文件压缩变小文件，&nbsp;&nbsp;&nbsp;从而减少网络传输，但是增加了cpu的计算负载。</p>
<p><strong>　　Spark</strong>里面io也是不可避免的，但是网络传输spark里面进行了优化：</p>
<p>　　Spark把rdd进行分区（分片），放在集群上并行计算。同一个rdd分片100个，10个节点，平均一个节点10个分区，当进行sum型的计算的时候，先进行每个分区的sum，然后把sum值shuffle传输到主程序进行全局sum，所以进行sum型计算对网络传输非常小。但对于进行join型的计算的时候，需要把数据本身进行shuffle，网络开销很大。</p>
<p>spark是如何优化这个问题的呢？</p>
<p>　　Spark把key－value rdd通过key的hashcode进行分区，而且保证相同的key存储在同一个节点上，这样对改rdd进行key聚合时，就不需要shuffle过程，我们进行mapreduce计算的时候为什么要进行shuffle？，就是说mapreduce里面网络传输主要在shuffle阶段，<strong>shuffle的根本原因是相同的key存在不同的节点上，按key进行聚合的时候不得不进行shuffle</strong>。shuffle是非常影响网络的，它要把所有的数据混在一起走网络，然后它才能把相同的key走到一起。<strong>要</strong><strong>进行</strong><strong>shuffle是存储决定的。</strong></p>
<p>　　Spark从这个教训中得到启发，spark会把key进行分区，也就是key的hashcode进行分区，相同的key，hashcode肯定是一样的，所以它进行分区的时候100t的数据分成10分，每部分10个t，它能确保相同的key肯定在一个分区里面，而且它能保证存储的时候相同的key能够存在同一个节点上。比如一个rdd分成了100份，集群有10个节点，所以每个节点存10份，每一分称为每个分区，spark能保证相同的key存在同一个节点上，实际上相同的key存在同一个分区。</p>
<p>　　key的分布不均决定了有的分区大有的分区小。没法分区保证完全相等，但它会保证在一个接近的范围。所以mapreduce里面做的某些工作里边，spark就不需要shuffle了，spark解决网络传输这块的根本原理就是这个。</p>
<p>　　进行join的时候是两个表，不可能把两个表都分区好，通常情况下是把用的频繁的大表事先进行分区，小表进行关联它的时候小表进行shuffle过程。</p>
<p>　　大表不需要shuffle。</p>
<p>　　需要在工作节点间进行数据混洗的转换极大地受益于分区。这样的转换是&nbsp;&nbsp;cogroup，groupWith，join，leftOuterJoin，rightOuterJoin，groupByKey，reduceByKey，combineByKey&nbsp;和lookup。</p>
<div>　　<strong>分区是可配置的，只要RDD是基于键值对的即可</strong>。</div>
<h2>三、Spark分区原则及方法</h2>
<p>RDD分区的一个<strong>分区原则：尽可能是得分区的个数等于集群核心数目</strong></p>
<p>无论是本地模式、Standalone模式、YARN模式或Mesos模式，我们都可以<strong><em>通过spark.default.parallelism来配置其默认分区个数</em></strong>，若没有设置该值，则根据不同的集群环境确定该值</p>
<h3>3.1　本地模式</h3>
<h4>（1）默认方式</h4>
<p>以下这种默认方式就一个分区</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503184117132-933712151.png" alt="" width="700" /></p>
<p>结果</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503184211834-340940238.png" alt="" width="700" /></p>
<h4>（2）手动设置</h4>
<p>设置了几个分区就是几个分区</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503184315304-1438737967.png" alt="" width="700" /></p>
<p>结果</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503184417483-992690743.png" alt="" width="700" /></p>
<h4>（3）跟local[n] 有关</h4>
<p>n等于几默认就是几个分区</p>
<p>如果n=* 那么分区个数就等于cpu core的个数</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503184602132-1762283216.png" alt="" width="700" /></p>
<p>结果</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503184628115-1042310352.png" alt="" width="700" /></p>
<p>本机电脑查看cpu core，我的电脑--》右键管理--》设备管理器--》处理器</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503184724600-1103843046.png" alt="" width="700" /></p>
<h4>（4）参数控制</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503185007050-446009891.png" alt="" width="700" /></p>
<p>结果</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503185028293-1238150539.png" alt="" width="700" /></p>
<h3>3.2　YARN模式</h3>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503190552013-885991110.png" alt="" width="700" /></p>
<p>&nbsp;进入defaultParallelism方法</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503190651046-910979790.png" alt="" /></p>
<p>继续进入defaultParallelism方法</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503190749812-1860089737.png" alt="" width="500" /></p>
<p>这个一个trait，其实现类是（Ctrl+h）</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503190853986-1549098382.png" alt="" width="500" /></p>
<p>进入TaskSchedulerImpl类找到defaultParallelism方法</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503190937719-848606303.png" alt="" width="700" /></p>
<p>继续进入defaultParallelism方法，又是一个trait，看其实现类</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503191109902-1222100636.png" alt="" width="700" /></p>
<p>Ctrl+h看SchedulerBackend类的实现类</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503191213525-716329324.png" alt="" width="500" /></p>
<p>进入CoarseGrainedSchedulerBackend找到defaultParallelism</p>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201805/1228818-20180503191320589-992961865.png" alt="" width="700" /></p>
<p><strong>totalCoreCount.get()是所有executor使用的core总数，和2比较去较大值</strong></p>
<p><strong>如果正常的情况下，那你设置了多少就是多少</strong></p>
<h2>四、分区器</h2>
<p>（1）如果是从HDFS里面读取出来的数据，不需要分区器。因为HDFS本来就分好区了。</p>
<p>　　&nbsp; 分区数我们是可以控制的，但是没必要有分区器。</p>
<p>（2）非key-value RDD分区，没必要设置分区器</p>
<div class="cnblogs_code">
<pre>al testRDD = sc.textFile("C:\\Users\\Administrator\\IdeaProjects\\myspark\\src\\main\\hello.txt")
  .flatMap(line =&gt; line.split(","))
  .map(word =&gt; (word, 1)).partitionBy(new HashPartitioner(2))</pre>
</div>
<pre><strong>　　没必要设置，但是非要设置也行。</strong></pre>
<p>（3）Key-value形式的时候，我们就有必要了。</p>
<p><strong>HashPartitioner</strong></p>
<div class="cnblogs_code">
<pre>val resultRDD = testRDD.reduceByKey(new HashPartitioner(2),(x:Int,y:Int) =&gt; x+ y)
//如果不设置默认也是HashPartitoiner，分区数跟spark.default.parallelism一样
println(resultRDD.partitioner)
println("resultRDD"+resultRDD.getNumPartitions)</pre>
</div>
<p><strong>RangePartitioner</strong></p>
<div class="cnblogs_code">
<pre>val resultRDD = testRDD.reduceByKey((x:Int,y:Int) =&gt; x+ y)
val newresultRDD=resultRDD.partitionBy(new RangePartitioner[String,Int](3,resultRDD))
println(newresultRDD.partitioner)
println("newresultRDD"+newresultRDD.getNumPartitions)</pre>
</div>
<pre><strong>注：按照范围进行分区的，如果是字符串，那么就按字典顺序的范围划分。如果是数字，就按数据自的范围划分</strong>。</pre>
<p><strong>自定义分区</strong></p>
<p><strong>需要实现2个方法</strong></p>
<div class="cnblogs_code">
<pre>class MyPartitoiner(val numParts:Int) extends  Partitioner{
  override def numPartitions: Int = numParts
  override def getPartition(key: Any): Int = {
    val domain = new URL(key.toString).getHost
    val code = (domain.hashCode % numParts)
    if (code &lt; 0) {
      code + numParts
    } else {
      code
    }
  }
}

object DomainNamePartitioner {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("word count").setMaster("local")

    val sc = new SparkContext(conf)

    val urlRDD = sc.makeRDD(Seq(("http://baidu.com/test", 2),
      ("http://baidu.com/index", 2), ("http://ali.com", 3), ("http://baidu.com/tmmmm", 4),
      ("http://baidu.com/test", 4)))
    //Array[Array[(String, Int)]]
    // = Array(Array(),
    // Array((http://baidu.com/index,2), (http://baidu.com/tmmmm,4),
    // (http://baidu.com/test,4), (http://baidu.com/test,2), (http://ali.com,3)))
    val hashPartitionedRDD = urlRDD.partitionBy(new HashPartitioner(2))
    hashPartitionedRDD.glom().collect()

    //使用spark-shell --jar的方式将这个partitioner所在的jar包引进去，然后测试下面的代码
    // spark-shell --master spark://master:7077 --jars spark-rdd-1.0-SNAPSHOT.jar
    val partitionedRDD = urlRDD.partitionBy(new MyPartitoiner(2))
    val array = partitionedRDD.glom().collect()

  }
}</pre>
</div></div>

</body>
</html>
